package com.flink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;

public class CountWindow {

    protected static void keyBy(StreamExecutionEnvironment environment) {
        DataStreamSource<String> socketTextStream = environment.socketTextStream("localhost", 1234);

        SingleOutputStreamOperator<Tuple2<String, String>> mapOperator = socketTextStream.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String originalData) throws Exception {
                return Tuple2.of(originalData, originalData);
            }
        });

        KeyedStream<Tuple2<String, String>, String> tuple2StringKeyedStream = mapOperator.keyBy(new KeySelector<Tuple2<String, String>, String>() {
            @Override
            public String getKey(Tuple2<String, String> stringStringTuple2) throws Exception {
                return stringStringTuple2.f0;
            }
        });

        SingleOutputStreamOperator<String> mapFunction = tuple2StringKeyedStream.map(new RichMapFunction<Tuple2<String, String>, String>() {
            @Override
            public String map(Tuple2<String, String> stringStringTuple2) throws Exception {
                System.out.println(stringStringTuple2.f0 + "   " + getRuntimeContext().getIndexOfThisSubtask());
                return stringStringTuple2.f0;
            }
        });

        mapFunction.print();
    }

    protected static void singlePreCountWindowAll(StreamExecutionEnvironment environment) {
        DataStreamSource<String> socketTextStream = environment.socketTextStream("localhost", 1234);

        SingleOutputStreamOperator<Tuple2<String, Integer>> mapOperator = socketTextStream.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String originalData) throws Exception {
                String[] split = originalData.split(",");
                return Tuple2.of(split[0], 1);
            }
        });
        AllWindowedStream<Tuple2<String, Integer>, GlobalWindow> tuple2GlobalWindowAllWindowedStream = mapOperator.countWindowAll(5);

        tuple2GlobalWindowAllWindowedStream.sum(1).print();
    }

    /**
     * keyBy -> countWindow
     * 各个key单独计算不会互相干扰
     *
     * @param environment
     */
    protected static void singlePreKeyByCountWindow(StreamExecutionEnvironment environment) {
        DataStreamSource<String> socketTextStream = environment.socketTextStream("localhost", 1234);

        SingleOutputStreamOperator<Tuple2<String, Integer>> mapOperator = socketTextStream.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String originalData) throws Exception {
                String[] split = originalData.split(",");
                return Tuple2.of(split[0], 1);
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyedOperator = mapOperator.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> stringStringTuple2) throws Exception {
                return stringStringTuple2.f0;
            }
        });

        WindowedStream<Tuple2<String, Integer>, String, GlobalWindow> countWindow = keyedOperator.countWindow(5);

        countWindow.sum(1).print();
    }

}
